package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeTrademarkCategoryUserRefundBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.HBaseUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.log4j.net.JMSAppender;

import java.time.Duration;
import java.util.HashSet;
import java.util.Iterator;

public class DwsTradeTrademarkCategoryUserRefundWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeTrademarkCategoryUserRefundWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_REFUND,"dws_trade_trademark_category_user_refund_window",4,10031);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        //etl
//        dataStreamSource.print();
        SingleOutputStreamOperator<JSONObject> eltedStream = eltStream(dataStreamSource);


        // join 补齐字段
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> process = getTrademarkCategoryUserRefundBeanSingleOutputStreamOperator(eltedStream);

        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> mapstream = getTradeTrademarkCategoryUserRefundBeanSingleOutputStreamOperator(process);

//        mapstream.print();

        //水位线
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> waterStream = waterStream(mapstream);

        // 分组 开窗 聚合

        // bean.getUserId() + "_" + bean.getCategory3Id() + "_" + bean.getTrademarkId()

        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> streamOperator = reduce_WinStream(waterStream);
        streamOperator.print();


        //写入 doris
        streamOperator.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks("dws_trade_trademark_category_user_refund_window"));

    }

    private static  SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean>  reduce_WinStream(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> waterStream) {
       return waterStream.keyBy((x -> x.getUserId() + "-" + x.getCategory3Id() + "-" + x.getTrademarkId()))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeTrademarkCategoryUserRefundBean>() {
                    @Override
                    public TradeTrademarkCategoryUserRefundBean reduce(TradeTrademarkCategoryUserRefundBean t1, TradeTrademarkCategoryUserRefundBean t2) throws Exception {
                        t1.setRefundCount(t1.getRefundCount() + t2.getRefundCount());

                        return t1;
                    }
                }, new WindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeTrademarkCategoryUserRefundBean> iterable, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {
                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<TradeTrademarkCategoryUserRefundBean> iterator = iterable.iterator();

                        TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean = iterator.next();

                        tradeTrademarkCategoryUserRefundBean.setStt(DateFormatUtil.tsToDateTime(start));
                        tradeTrademarkCategoryUserRefundBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        tradeTrademarkCategoryUserRefundBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));

                        collector.collect(tradeTrademarkCategoryUserRefundBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getTrademarkCategoryUserRefundBeanSingleOutputStreamOperator(SingleOutputStreamOperator<JSONObject> eltedStream) {
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> process = eltedStream.process(new ProcessFunction<JSONObject, TradeTrademarkCategoryUserRefundBean>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TradeTrademarkCategoryUserRefundBean>.Context context, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {

                String skuId = jsonObject.getString("sku_id");
                Long ts = jsonObject.getLong("ts");
                HashSet<String> set = new HashSet<>();
                set.add(jsonObject.getString("order_id"));

                String tsToDate = DateFormatUtil.tsToDateTime(ts);


                collector.collect(TradeTrademarkCategoryUserRefundBean.builder()
                        .skuId(skuId)
                        .curDate(tsToDate)
                        .orderIdSet(set)
                        .ts(ts)
                                .refundCount(1L)
                        .userId(jsonObject.getString("user_id"))
                        .build());
            }
        });
        return process;
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getTradeTrademarkCategoryUserRefundBeanSingleOutputStreamOperator(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> process) {
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> mapstream = process.map(new RichMapFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean>() {
            private Connection connection;

            @Override
            public void open(Configuration parameters) throws Exception {
                connection = HBaseUtil.getHBaseConnection();
            }

            @Override
            public void close() throws Exception {
                connection.close();
            }

            @Override
            public TradeTrademarkCategoryUserRefundBean map(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean) throws Exception {
                //根据 sku_id 获取 tm_id，category3_id

                String skuId = tradeTrademarkCategoryUserRefundBean.getSkuId();

                JSONObject dimSkuInfo = HBaseUtil.getCells(connection, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId);

                String tmId = dimSkuInfo.getString("tm_id");
                String category3Id = dimSkuInfo.getString("category3_id");


                //tmId -> tm_name
                JSONObject dimBaseTrademark = HBaseUtil.getCells(connection, Constant.HBASE_NAMESPACE, "dim_base_trademark", tmId);

                String tmName = dimBaseTrademark.getString("tm_name");


                //category3Id   -> category2_id  name
                JSONObject dimBaseCategory3 = HBaseUtil.getCells(connection, Constant.HBASE_NAMESPACE, "dim_base_category3", category3Id);

                String dimBaseCategory3name = dimBaseCategory3.getString("name");
                String category2Id = dimBaseCategory3.getString("category2_id");

                //category2Id -> name  category1_id
                JSONObject dimBaseCategory2 = HBaseUtil.getCells(connection, Constant.HBASE_NAMESPACE, "dim_base_category2", category2Id);

                String dimBaseCategory2name = dimBaseCategory2.getString("name");
                String category1Id = dimBaseCategory2.getString("category1_id");

                //category1Id -> name    dim_base_category1
                JSONObject dimBaseCategory1 = HBaseUtil.getCells(connection, Constant.HBASE_NAMESPACE, "dim_base_category1", category1Id);

                String dimBaseCategory1name = dimBaseCategory1.getString("name");

                tradeTrademarkCategoryUserRefundBean.setCategory1Id(category1Id);
                tradeTrademarkCategoryUserRefundBean.setCategory2Id(category2Id);
                tradeTrademarkCategoryUserRefundBean.setCategory3Id(category3Id);
                tradeTrademarkCategoryUserRefundBean.setCategory1Name(dimBaseCategory1name);
                tradeTrademarkCategoryUserRefundBean.setCategory2Name(dimBaseCategory2name);
                tradeTrademarkCategoryUserRefundBean.setCategory3Name(dimBaseCategory3name);
                tradeTrademarkCategoryUserRefundBean.setTrademarkId(tmId);
                tradeTrademarkCategoryUserRefundBean.setTrademarkName(tmName);


                return tradeTrademarkCategoryUserRefundBean;

            }
        });
        return mapstream;
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> waterStream(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> mapstream) {
         return mapstream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TradeTrademarkCategoryUserRefundBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getTs()).withIdleness(Duration.ofSeconds(1)));
    }

    private static  SingleOutputStreamOperator<JSONObject> eltStream(DataStreamSource<String> dataStreamSource) {
        return  dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);

                    if (jsonObject != null) {

                        Long ts = jsonObject.getLong("ts");

                        //tm_id，category3_id。
                        String skuId = jsonObject.getString("sku_id");
                        if (ts != null && skuId != null) {

                            jsonObject.put("ts", ts * 1000);
                            collector.collect(jsonObject);
                        }

                    }
                }
            }
        });
    }
}
